#include "head.h"

void pull_message(char *p) //mqtt发布
{
    pubmsg.payload = p;
    pubmsg.payloadlen = (int)strlen(p);
    pubmsg.qos = QOS;
    pubmsg.retained = 0;
    MQTTClient_publishMessage(client, TOPIC_UP, &pubmsg, &token);
    return;
}
void periodic_reporting() //整体上报
{
    char i_val[12] = "";
    char f_val[12] = "";
    struct std_node *p = (struct std_node *)(dev_num + 1);
    cJSON *root = cJSON_CreateObject();
    cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
    cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
    cJSON *data = cJSON_CreateArray();
    for (int i = 0; i < *dev_num; i++)
    {
        cJSON *item = cJSON_CreateObject();
        cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(p->key));
        switch (p->type)
        {
        case 1:
            if (p->new_val.b_val)
            {
                cJSON_AddItemToObject(item, "val", cJSON_CreateString("true"));
            }
            else
            {
                cJSON_AddItemToObject(item, "val", cJSON_CreateString("false"));
            }
            break;
        case 2:
            sprintf(i_val, "%d", p->new_val.i_val);
            cJSON_AddItemToObject(item, "val", cJSON_CreateString(i_val));
            break;
        case 3:
            sprintf(f_val, "%.2f", p->new_val.f_val);
            cJSON_AddItemToObject(item, "val", cJSON_CreateString(f_val));
            break;
        default:
            break;
        }
        cJSON_AddItemToObject(item, "type", cJSON_CreateNumber(p->type));
        cJSON_AddItemToArray(data, item);
        p++;
    }
    cJSON_AddItemToObject(root, "data", data);
    char *json_str = cJSON_Print(root);
    //调用mqtt的发布
    pull_message(json_str);
    //释放内存
    cJSON_Delete(root);
    free(json_str);
    json_str = NULL;
    return;
}
void change_reporting() //变化上报
{
    sleep(1);
    cJSON *root = cJSON_CreateObject();
    cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(1));
    cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
    cJSON *data = cJSON_CreateArray();
    char i_val[12] = "";
    char f_val[12] = "";
    struct std_node *p = (struct std_node *)(dev_num + 1);
    for (int i = 0; i < *dev_num; i++)
    {
        switch (p->type)
        {
        case 1:
            if (p->new_val.b_val != p->old_val.b_val)
            {
                cJSON *item = cJSON_CreateObject();
                cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(p->key));
                if (p->new_val.b_val)
                {
                    cJSON_AddItemToObject(item, "val", cJSON_CreateString("true"));
                }
                else
                {
                    cJSON_AddItemToObject(item, "val", cJSON_CreateString("false"));
                }
                cJSON_AddItemToObject(item, "type", cJSON_CreateNumber(p->type));
                cJSON_AddItemToArray(data, item);
            }
            break;
        case 2:
            if (p->new_val.i_val != p->old_val.i_val)
            {
                cJSON *item = cJSON_CreateObject();
                cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(p->key));
                sprintf(i_val, "%d", p->new_val.i_val);
                cJSON_AddItemToObject(item, "val", cJSON_CreateString(i_val));
                cJSON_AddItemToObject(item, "type", cJSON_CreateNumber(p->type));
                cJSON_AddItemToArray(data, item);
            }
            break;
        case 3:
            if (p->new_val.f_val != p->old_val.f_val)
            {
                cJSON *item = cJSON_CreateObject();
                cJSON_AddItemToObject(item, "key", cJSON_CreateNumber(p->key));
                sprintf(f_val, "%.2f", p->new_val.f_val);
                cJSON_AddItemToObject(item, "val", cJSON_CreateString(f_val));
                cJSON_AddItemToObject(item, "type", cJSON_CreateNumber(p->type));
                cJSON_AddItemToArray(data, item);
            }
            break;
        default:
            break;
        }
        p++;
    }
    cJSON_AddItemToObject(root, "data", data);
    int datalen = cJSON_GetArraySize(data);
    if (datalen)
    {
        char *json_str = cJSON_Print(root);
        printf("%s\n", json_str);
        //调用mqtt的发布
        pull_message(json_str);
        free(json_str);
        json_str = NULL;
    }
    //释放内存
    cJSON_Delete(root);
    return;
}

void historical_data_reporting(char *str) //历史数据上报
{
    sqlite3_stmt *stmt;

    cJSON *ctl_root = cJSON_Parse(str);
    cJSON *ctl_data = cJSON_GetObjectItem(ctl_root, "data");
    cJSON *ctl_key = cJSON_GetObjectItem(ctl_data, "key");
    cJSON *ctl_limit = cJSON_GetObjectItem(ctl_data, "limit");

    cJSON *root = cJSON_CreateObject();
    cJSON_AddItemToObject(root, "type", cJSON_CreateNumber(4));
    cJSON_AddItemToObject(root, "result", cJSON_CreateNumber(0));
    cJSON *data = cJSON_CreateArray();
    cJSON_AddItemToObject(root, "data", data); // 将数组添加到root对象中
    cJSON *timestamp1 = cJSON_GetArrayItem(ctl_limit, 0);
    cJSON *timestamp2 = cJSON_GetArrayItem(ctl_limit, 1);
    char sqlstr[1024];
    char *errmsg;

    // 准备 SQL 语句
    const char *sql = "SELECT * FROM workshop WHERE key=? AND time BETWEEN ? AND ?;";
    rc = sqlite3_prepare_v2(db, sql, -1, &stmt, NULL);
    if (rc)
    {
        fprintf(stderr, "SQL 准备失败: %s\n", sqlite3_errmsg(db));
        return;
    }
    else
    {
        fprintf(stderr, "SQL 准备成功\n");
    }
    // 绑定参数
    sqlite3_bind_int(stmt, 1, ctl_key->valueint);
    sqlite3_bind_int(stmt, 2, timestamp1->valueint);
    sqlite3_bind_int(stmt, 3, timestamp2->valueint);

    // 执行查询
    while (sqlite3_step(stmt) == SQLITE_ROW)
    {
        cJSON *item = cJSON_CreateObject();
        cJSON_AddItemToObject(item, "time", cJSON_CreateNumber(sqlite3_column_int(stmt, 0)));
        cJSON_AddItemToObject(item, "val", cJSON_CreateString(sqlite3_column_text(stmt, 2)));
        cJSON_AddItemToArray(data, item);
    }
    // 清理
    sqlite3_finalize(stmt);

    int datalen = cJSON_GetArraySize(data);
    if (datalen)
    {
        char *json_str = cJSON_Print(root);
        printf("%s\n", json_str);
        //调用mqtt的发布
        pull_message(json_str);
        free(json_str);
        json_str = NULL;
    }
    else
    {
        printf("没有数据，拼不上啊!!!!!!\n");
    }

    // 清理root（如果你不再需要它）
    cJSON_Delete(ctl_root);
    cJSON_Delete(root);
    return;
}

void send_ctl(char *str) //控制命令
{
    key_t msg_key;
    int msgid;
    if ((msg_key = ftok("/mnt/msg.txt", 'b')) < 0)
    {
        perror("ftok err");
        return;
    }

    //打开或创建消息队列
    msgid = msgget(msg_key, IPC_CREAT | IPC_EXCL | 0666); //创建消息队列，如果之前有就返回-1
    if (msgid <= 0)
    {
        if (errno == EEXIST)
            msgid = msgget(msg_key, 0666); //直接打开消息队列
        else
        {
            perror("msgget err");
            return;
        }
    }

    //添加消息
    struct msgbuf send_msg;
    cJSON *root = cJSON_Parse(str);
    cJSON *data = cJSON_GetObjectItem(root, "data");
    cJSON *key = cJSON_GetObjectItem(data, "key");
    cJSON *val = cJSON_GetObjectItem(data, "val");
    struct std_node *p = (struct std_node *)(dev_num + 1);
    for (int i = 0; i < *dev_num; i++)
    {
        if (p->key == key->valueint)
        {
            send_msg.key = key->valueint;
            if (p->dev_type == 888)
            {
                send_msg.type = 10;
            }
            else if (p->dev_type == 666)
            {
                send_msg.type = 20;
            }
            switch (p->type)
            {
            case 1:
                send_msg.val.b_val = atoi(val->valuestring);
                break;
            case 2:
                send_msg.val.i_val = atoi(val->valuestring);
                break;
            case 3:
                send_msg.val.f_val = strtod(val->valuestring, NULL);
                break;
            default:
                break;
            }
        }
        p++;
    }
    if (msgsnd(msgid, &send_msg, sizeof(send_msg) - sizeof(long), 0) == -1)
    {
        printf("!!!发送消息队列控制命令失败!!!\n");
    }
}

void mode_modification(char *str) //模式修改
{
    cJSON *sroot = cJSON_Parse(str);
    cJSON *data = cJSON_GetObjectItem(sroot, "data");
    cJSON *type = cJSON_GetObjectItem(data, "type");
    int time_s = 0;
    if (type->valueint == 2)
    {
        cJSON *period = cJSON_GetObjectItem(data, "period");
        time_s = period->valueint;
    }
    // 1. 读取JSON文件
    FILE *file = fopen("/mnt/node.json", "r");
    char buffer[65536];
    fread(buffer, 1, sizeof(buffer) - 1, file);
    fclose(file);
    buffer[sizeof(buffer) - 1] = 0; // 确保字符串以null终止

    // 2. 解析JSON
    cJSON *root = cJSON_Parse(buffer);

    // 假设我们要修改一个名为"name"的字段的值
    cJSON *report = cJSON_GetObjectItem(root, "report");
    cJSON_ReplaceItemInObject(report, "type", cJSON_CreateNumber(type->valueint));
    cJSON_ReplaceItemInObject(report, "period", cJSON_CreateNumber(time_s));

    // 3. 将修改后的数据写回到文件
    char *output = cJSON_Print(root);
    file = fopen("/mnt/node.json", "w");
    fputs(output, file);
    fclose(file);

    // 4. 清理
    cJSON_Delete(root);
    cJSON_Delete(sroot);
    free(output);
}

int init_sql() //初始化数据库
{
    // 打开数据库
    if (sqlite3_open("./workshop.db", &db) != SQLITE_OK)
    {
        fprintf(stderr, "open workshop err %s", sqlite3_errmsg(db));
        return -1;
    }
    //操作数据库
    //创建表
    if (sqlite3_exec(db, "create table if not exists workshop(time INTEGER NOT NULL,key INTEGER NOT NULL,value TEXT,PRIMARY KEY(time,key));",
                     NULL, NULL, &errmsg) != SQLITE_OK)
    {
        fprintf(stderr, "create workshop err %s\n", errmsg);
        return -1;
    }
    return 0;
}

void *reporting_thread(void *arg) //上报线程
{
    while (1)
    {
        switch (send_type)
        {
        case 1: //变化上报
            change_reporting();
            break;
        case 2: //定时上报
            periodic_reporting();
            sleep(send_time);
            break;
        default: //刷新上报
            break;
        }
    }
    return NULL;
}

void *data_storage(void *arg) //数据存储线程
{
    while (1)
    {
        sleep(60); //等待1分钟存储一次
        time_t t;
        time(&t); //当前时间一直改变
        char b_val[12] = "";
        char i_val[12] = "";
        char f_val[12] = "";
        char sql[128] = "";
        struct std_node *p = (struct std_node *)(dev_num + 1);
        for (int i = 0; i < *dev_num; i++)
        {
            switch (p->type)
            {
            case 1:
                sprintf(b_val, "%d", p->new_val.b_val);
                sprintf(sql, "insert into workshop values(%ld,%d,%s);", t, p->key, b_val);
                printf("!!!!!!!!!!!!!b_val!!!!!!!!!ok!!!!!!!\n");
                break;
            case 2:
                sprintf(i_val, "%d", p->new_val.i_val);
                sprintf(sql, "insert into workshop values(%ld,%d,%s);", t, p->key, i_val);
                printf("!!!!!!!!!!!!!i_val!!!!!!!!!ok!!!!!!!\n");
                break;
            case 3:
                sprintf(f_val, "%f", p->new_val.f_val);
                sprintf(sql, "insert into workshop values(%ld,%d,%s);", t, p->key, f_val);
                printf("!!!!!!!!!!!!!f_val!!!!!!!!!ok!!!!!!!\n");
                break;
            default:
                break;
            }
            //插入数据
            if (sqlite3_exec(db, sql, NULL, NULL, &errmsg) != SQLITE_OK)
            {
                fprintf(stderr, "insert workshop erro %s\n", errmsg);
                return NULL;
            }
            p++;
        }
    }
}

int msgarrvd(void *context, char *topicName, int topicLen, MQTTClient_message *message)
{
    int i;
    char *payloadptr;

    printf("Message arrived\n");
    printf("     topic: %s\n", topicName);
    printf("   message: ");

    payloadptr = message->payload;

    cJSON *root = cJSON_Parse(payloadptr);
    if (root == NULL)
    {
        printf("下发数据不能为空\n");
        return -1;
    }
    cJSON *item = cJSON_GetObjectItem(root, "type");

    /*
        根据type的值来进行分别操作
        1：采集指令/采集回复指令
        2：控制指令/控制回复指令
        3：模式修改
        4：历史数据查询*/
    switch (item->valueint)
    {
    case 1:
        periodic_reporting();
        break;
    case 2:
        send_ctl(payloadptr);
        break;
    case 3:
        mode_modification(payloadptr);
        break;
    case 4:
        historical_data_reporting(payloadptr);
        break;
    default:
        break;
    }
    printf("%s\n", payloadptr);
    printf("-------%s----%s----%d----\n", __FILE__, __func__, __LINE__);

    cJSON_Delete(root);
    MQTTClient_freeMessage(&message);
    MQTTClient_free(topicName);
    return 1;
}

volatile MQTTClient_deliveryToken deliveredtoken;
void delivered(void *context, MQTTClient_deliveryToken dt)
{
    printf("Message with token value %d delivery confirmed\n", dt);
    deliveredtoken = dt;
}
void connlost(void *context, char *cause)
{
    printf("\nConnection lost\n");
    printf("     cause: %s\n", cause);
}

int main(int argc, char *argv[])
{
    //1.读点表
    char f_buf[2048] = "";    
    FILE *fp = NULL;
    while (1) //循环等待设备搜索进程下发数据点表
    {
        fp = fopen("/mnt/node.json", "r");
        if (fp != NULL)
        {
            printf("接受到数据点表\n");
            break;
        }
    }

    fread(f_buf, 1, sizeof(f_buf), fp);
    cJSON *root = cJSON_Parse(f_buf); //获取root根节点
    cJSON *smqtt_server = cJSON_GetObjectItem(root, "mqtt_server");
    cJSON *saddr = cJSON_GetObjectItem(smqtt_server, "addr");
    cJSON *sport = cJSON_GetObjectItem(smqtt_server, "port");
    sprintf(ADDRESS, "tcp://%s:%d", saddr->valuestring, sport->valueint);
    printf("%s\n", ADDRESS);

    //3.建立数据库
    if (init_sql() != 0)
    {
        perror("init database erro");
        return -1;
    }

    //4映射共享内存
    struct shm_param shm;
    shm_init(&shm, "shm.txt", sizeof(shm));
    dev_num = shm_getaddr(&shm);
    struct std_node *p = (struct std_node *)(dev_num + 1);

    //5.解析点表
    //5.1反序列化操作
    cJSON *report = cJSON_GetObjectItem(root, "report"); //获取上报模式
    cJSON *stype = cJSON_GetObjectItem(report, "type");  //获取上报模式的type
    send_type = stype->valueint;
    if (send_type == 2)
    {
        cJSON *period = cJSON_GetObjectItem(report, "period"); //获取周期上报时间
        send_time = period->valueint;
    }
    cJSON *modbus = cJSON_GetObjectItem(root, "modbus"); //从根节点上获取键叫"modbus"的子节点
    cJSON *m_data = cJSON_GetObjectItem(modbus, "data"); //从modbus节点上获取data数组
    mod_dev_num = cJSON_GetArraySize(m_data);            //获取modbus中的设备数
    cJSON *stm32 = cJSON_GetObjectItem(root, "stm32");   //从根节点上获取键叫"stm32"的子节点
    cJSON *s_data = cJSON_GetObjectItem(stm32, "data");  //从modbus节点上获取data数组
    stm_dev_num = cJSON_GetArraySize(s_data);            //获取stm32中的设备数
    *dev_num = stm_dev_num + mod_dev_num;                //设定设备总数
    cJSON *item;
    cJSON *key;
    cJSON *type;
    for (int i = 0; i < stm_dev_num; i++)
    {
        item = cJSON_GetArrayItem(s_data, i);
        key = cJSON_GetObjectItem(item, "key");
        p->key = key->valueint;
        type = cJSON_GetObjectItem(item, "type");
        p->type = type->valueint;
        p->dev_type = 666;
        p++;
    }
    for (int j = 0; j < mod_dev_num; j++)
    {
        item = cJSON_GetArrayItem(m_data, j);
        key = cJSON_GetObjectItem(item, "key");
        p->key = key->valueint;
        type = cJSON_GetObjectItem(item, "type");
        p->type = type->valueint;
        p->dev_type = 888;
        p++;
    }
    //释放root空间
    cJSON_Delete(root);

    //2.连接mqtt
    //2.1创建客户端
    MQTTClient_create(&client, ADDRESS, CLIENTID,
                      MQTTCLIENT_PERSISTENCE_NONE, NULL);
    conn_opts.keepAliveInterval = 20;
    conn_opts.cleansession = 1;
    MQTTClient_setCallbacks(client, NULL, connlost, msgarrvd, delivered);
    //2.2建立连接
    if ((rc = MQTTClient_connect(client, &conn_opts)) != MQTTCLIENT_SUCCESS)
    {
        printf("Failed to connect, return code %d\n", rc);
        exit(EXIT_FAILURE);
    }
    MQTTClient_subscribe(client, TOPIC_DOWN, QOS);

    //创建上报线程
    pthread_t up_tid;
    if (pthread_create(&up_tid, NULL, reporting_thread, NULL) != 0)
    {
        perror("create reporting thread erro!");
        return -1;
    }

    //创建数据存储线程
    pthread_t save_tid;
    if (pthread_create(&save_tid, NULL, data_storage, NULL) != 0)
    {
        perror("create data storage erro!");
        return -1;
    }
    pthread_join(save_tid, NULL);
    printf("-------%s----%s----%d----\n", __FILE__, __func__, __LINE__);
    pthread_join(up_tid, NULL);
    printf("-------%s----%s----%d----\n", __FILE__, __func__, __LINE__);

    // 关闭数据库
    sqlite3_close(db);
    //释放共享内存
    shm_del(&shm);
    //结束订阅
    MQTTClient_unsubscribe(client, TOPIC_DOWN);
    MQTTClient_disconnect(client, 10000);
    MQTTClient_destroy(&client);
    return rc;
}
